116a78094fd53fc13d3b613f422a9fb220b95f1f,hoodie-client/src/main/java/com/uber/hoodie/index/HoodieBloomIndex.java,HoodieBloomIndex,fetchRecordLocation,#JavaRDD#HoodieTable#,96

Before Change


    public JavaPairRDD<HoodieKey, Optional<String>> fetchRecordLocation(
        JavaRDD<HoodieKey> hoodieKeys, final HoodieTable<T> hoodieTable) {
        JavaPairRDD<String, String> partitionRecordKeyPairRDD =
            hoodieKeys.mapToPair(new PairFunction<HoodieKey, String, String>() {
                @Override
                public Tuple2<String, String> call(HoodieKey key) throws Exception {
                    return new Tuple2<>(key.getPartitionPath(), key.getRecordKey());
                }
            });

        // Lookup indexes for all the partition/recordkey pair
        JavaPairRDD<String, String> rowKeyFilenamePairRDD =
            lookupIndex(partitionRecordKeyPairRDD, hoodieTable);

        JavaPairRDD<String, HoodieKey> rowKeyHoodieKeyPairRDD =
            hoodieKeys.mapToPair(new PairFunction<HoodieKey, String, HoodieKey>() {
                @Override
                public Tuple2<String, HoodieKey> call(HoodieKey key) throws Exception {
                    return new Tuple2<>(key.getRecordKey(), key);
                }
            });

        return rowKeyHoodieKeyPairRDD.leftOuterJoin(rowKeyFilenamePairRDD).mapToPair(
        new PairFunction<Tuple2<String, Tuple2<HoodieKey, org.apache.spark.api.java.Optional<String>>>, HoodieKey, Optional<String>>() {
                @Override
                public Tuple2<HoodieKey, Optional<String>> call(
                    Tuple2<String, Tuple2<HoodieKey, org.apache.spark.api.java.Optional<String>>> keyPathTuple)
                    throws Exception {
                    Optional<String> recordLocationPath;
                    if (keyPathTuple._2._2.isPresent()) {
                        String fileName = keyPathTuple._2._2.get();
                        String partitionPath = keyPathTuple._2._1.getPartitionPath();
                        recordLocationPath = Optional.of(new Path(
                            new Path(hoodieTable.getMetaClient().getBasePath(), partitionPath),
                            fileName).toUri().getPath());
                    } else {
                        recordLocationPath = Optional.absent();
                    }
                    return new Tuple2<>(keyPathTuple._2._1, recordLocationPath);
                }
            });
    }

    /**

After Change


                hoodieKeys.mapToPair(key -> new Tuple2<>(key.getRecordKey(), key));

        return rowKeyHoodieKeyPairRDD.leftOuterJoin(rowKeyFilenamePairRDD)
                .mapToPair(keyPathTuple -> {
                    Optional<String> recordLocationPath;
                    if (keyPathTuple._2._2.isPresent()) {
                        String fileName = keyPathTuple._2._2.get();
                        String partitionPath = keyPathTuple._2._1.getPartitionPath();
                        recordLocationPath = Optional.of(new Path(
                                new Path(hoodieTable.getMetaClient().getBasePath(), partitionPath),
                                fileName).toUri().getPath());
                    } else {
                        recordLocationPath = Optional.absent();
                    }
                    return new Tuple2<>(keyPathTuple._2._1, recordLocationPath);
                });
    }

    /**